package com.zhu.app.dwd;

import com.zhu.utils.MySqlUtil;
import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * 用户使用优惠券支付时，优惠券领用表的 used_time 字段会更新为支付时间，因此优
 * 惠券支付数据应满足两个条件：（1）操作类型为 update；（2）修改了 used_time 字段。
 * 使用优惠券支付后，优惠券领用表数据就不会再发生变化，所以在操作类型为 update 的前
 * 提下，只要 used_time 不为 null，就可以断定本次操作修改的是 used_time 字段。
 */
public class DWDToolCouponPayApp {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(1);

        //check point
           /*
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(ClusterParametersConfig.HDFS_CHECKPOINT_FILE_DIR);  //检查点保存在hdfs
        System.setProperty("HADOOP_USER_NAME", "zhu");
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);  //TimeOut
        streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);  //最大共存检查点
        streamExecutionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5 * 1000L));  //重启策略
        */

        StreamTableEnvironment streamTableEnvironment  = StreamTableEnvironment.create(streamExecutionEnvironment);
        Configuration configuration = streamTableEnvironment.getConfig().getConfiguration();
        configuration.setString("table.exec.state.ttl","5 s");

        //todo topic_db
        streamTableEnvironment.executeSql(ZhuKafkaUtil.getTopicDB("dwd_tool_coupon_pay"));

        // TODO 4. 读取优惠券领用表数据，筛选优惠券使用（支付）数据
        Table couponUsePay = streamTableEnvironment.sqlQuery("select\n" +
                        "data['id'] id,\n" +
                        "data['coupon_id'] coupon_id,\n" +
                        "data['user_id'] user_id,\n" +
                        "data['order_id'] order_id,\n" +
                        "date_format(data['used_time'],'yyyy-MM-dd') date_id,\n" +
                        "data['used_time'] used_time,\n" +
                        "`old` \n" +
                        "from topic_db\n" +
                        "where `table` = 'coupon_use'\n" +
                        "and `type` = 'update'\n" +
                        "and data['used_time'] is not null");
        streamTableEnvironment.createTemporaryView("coupon_use_pay", couponUsePay);

        // TODO 5. 建立 Kafka-Connector dwd_tool_coupon_order 表
        streamTableEnvironment.executeSql("create table dwd_tool_coupon_pay(\n" +
                "id string,\n" +
                "coupon_id string,\n" +
                "user_id string,\n" +
                "order_id string,\n" +
                "date_id string,\n" +
                "payment_time string \n" +
                ")" +
                ZhuKafkaUtil.getKafkaSinkDDL("dwd_tool_coupon_pay"));
// TODO 6. 将数据写入 Kafka-Connector 表
        streamTableEnvironment.executeSql("" +
                "insert into dwd_tool_coupon_pay select " +
                "id,\n" +
                "coupon_id,\n" +
                "user_id,\n" +
                "order_id,\n" +
                "date_id,\n" +
                "used_time payment_time\n" +
                " from coupon_use_pay").print();

        //todo mysql base_dic

        streamExecutionEnvironment.execute();
    }
}
